package edu.hawaii.ics.pjdb.worker.command_runners;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import edu.hawaii.ics.pjdb.result_transport.TupleServiceHelper;
import edu.hawaii.ics.pjdb.types.CsvPartitionCommand;
import edu.hawaii.ics.pjdb.types.Tuple;
import edu.hawaii.ics.pjdb.types.WorkerCommand;
import edu.hawaii.ics.pjdb.util.IOHelper;

/**
 * Reads data from a CSV file, partitions it based on a dynamically loaded partitioning function,
 * and sends the partitioned tuples to the output streams.
 * @author Christopher Foo
 *
 */
public class CsvPartitionRunner extends CommandRunner {
    private static final Logger logger = LoggerFactory
            .getLogger(CsvPartitionRunner.class);

    public CsvPartitionRunner(Connection localDb) {
        super(localDb);
    }

    @Override
    public void execute(WorkerCommand cmd) throws TException, SQLException {
        CsvPartitionCommand partitionCmd = cmd.getCommand()
                .getCsvPartitionCmd();

        // Read CSV Values
        List<Tuple> buffer = IOHelper.readCsvFile(partitionCmd.getCsvFile(),
                partitionCmd.isHeader(), logger);

        if (buffer == null) {
            return;
        }

        // Set up reflection stuff
        ClassLoader loader = ClassLoader.getSystemClassLoader();
        Class<?> myClass = null;
        try {
            myClass = loader
                    .loadClass("edu.hawaii.ics.pjdb.partition_function."
                            + partitionCmd.getPartitionFunctionName());
        } catch (ClassNotFoundException e) {
            System.err.println("Dynamic class loading exception: "
                    + e.getMessage());
            return;
        }
        Constructor<?> constructor = null;
        try {
            constructor = myClass.getConstructor();
        } catch (SecurityException e) {
            System.err.println("Dynamic class loading exception: "
                    + e.getMessage());
            return;
        } catch (NoSuchMethodException e) {
            System.err.println("Dynamic class loading exception: "
                    + e.getMessage());
            return;
        }
        Object partitionFunction = null;
        try {
            partitionFunction = constructor.newInstance();
        } catch (IllegalArgumentException e) {
            System.err.println("Dynamic class loading exception: "
                    + e.getMessage());
            return;
        } catch (InstantiationException e) {
            System.err.println("Dynamic class loading exception: "
                    + e.getMessage());
            return;
        } catch (IllegalAccessException e) {
            System.err.println("Dynamic class loading exception: "
                    + e.getMessage());
            return;
        } catch (InvocationTargetException e) {
            System.err.println("Dynamic class loading exception: "
                    + e.getMessage());
            return;
        }
        Method partition = null;
        try {
            partition = myClass.getMethod("partition", List.class, List.class,
                    Integer.class);
        } catch (SecurityException e) {
            System.err.println("Dynamic class loading exception: "
                    + e.getMessage());
            return;
        } catch (NoSuchMethodException e) {
            System.err.println("Dynamic class loading exception: "
                    + e.getMessage());
            return;
        }

        // Partition Tuples
        ArrayList<ArrayList<Tuple>> partitionBuffers;
        try {
            partitionBuffers = (ArrayList<ArrayList<Tuple>>) partition.invoke(
                    partitionFunction, buffer, partitionCmd.getArguments(),
                    partitionCmd.getPartitionFieldIndex());
            
            // If more partitions than outputs, collapse partitions.
            int numPartitions = partitionBuffers.size();
            int numOutputs = partitionCmd.getOutputAddressesSize();
            if(numPartitions > numOutputs) {     
                for(int i = 0, j = numOutputs; j < numPartitions; i++, j++) {
                    partitionBuffers.get(i % numOutputs).addAll(partitionBuffers.get(j));
                }
                
                for(int i = numPartitions - 1; i >= numOutputs; i--) {
                    partitionBuffers.remove(i);
                }
            }

            // If fewer partitions than outputs, add empty partitions
            while(partitionBuffers.size() < numOutputs) {
                partitionBuffers.add(new ArrayList<Tuple>());
            }
            
            // Send off to DbSync commands
            numPartitions = partitionBuffers.size();
            for (int i = 0; i < numPartitions; i++) {
                TupleServiceHelper.submitTuples(partitionCmd
                        .getOutputAddresses().get(i), partitionBuffers.get(i));
            }

        } catch (IllegalArgumentException e) {
            System.err.println("Dynamic class loading exception: "
                    + e.getMessage());
            return;
        } catch (IllegalAccessException e) {
            System.err.println("Dynamic class loading exception: "
                    + e.getMessage());
            return;
        } catch (InvocationTargetException e) {
            System.err.println(e.getTargetException().getMessage());
            return;
        }

    }

}
